/*
 *  Licensed to the Apache Software Foundation (ASF) under one or more
 *  contributor license agreements.  See the NOTICE file distributed with
 *  this work for additional information regarding copyright ownership.
 *  The ASF licenses this file to You under the Apache License, Version 2.0
 *  (the "License"); you may not use this file except in compliance with
 *  the License.  You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 *  Unless required by applicable law or agreed to in writing, software
 *  distributed under the License is distributed on an "AS IS" BASIS,
 *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 *  See the License for the specific language governing permissions and
 *  limitations under the License.
 */

package org.apache.jk.common;

import java.io.IOException;

import org.apache.coyote.InputBuffer;
import org.apache.coyote.OutputBuffer;
import org.apache.coyote.Request;
import org.apache.coyote.Response;
import org.apache.jk.core.Msg;
import org.apache.jk.core.MsgContext;
import org.apache.tomcat.util.buf.ByteChunk;
import org.apache.tomcat.util.buf.C2BConverter;
import org.apache.tomcat.util.buf.MessageBytes;
import org.apache.tomcat.util.http.HttpMessages;
import org.apache.tomcat.util.http.MimeHeaders;

/**
 * Generic input stream impl on top of ajp
 */
public class JkInputStream implements InputBuffer, OutputBuffer {
	private static org.apache.juli.logging.Log log = org.apache.juli.logging.LogFactory
			.getLog(JkInputStream.class);

	private Msg bodyMsg;
	private Msg outputMsg;
	private MsgContext mc;

	// Holds incoming chunks of request body data
	private MessageBytes bodyBuff = MessageBytes.newInstance();
	private MessageBytes tempMB = MessageBytes.newInstance();
	private boolean end_of_stream = false;
	private boolean isEmpty = true;
	private boolean isFirst = true;
	private boolean isReplay = false;
	private boolean isReadRequired = false;

	static {
		// Make certain HttpMessages is loaded for SecurityManager
		try {
			Class.forName("org.apache.tomcat.util.http.HttpMessages");
		} catch (Exception ex) {
			// ignore
		}
	}

	public JkInputStream(MsgContext context, int bsize) {
		mc = context;
		bodyMsg = new MsgAjp(bsize);
		outputMsg = new MsgAjp(bsize);
	}

	/**
	 * @deprecated
	 */
	public JkInputStream(MsgContext context) {
		this(context, 8 * 1024);
	}

	// -------------------- Jk specific methods --------------------

	/**
	 * Set the flag saying that the server is sending a body
	 */
	public void setIsReadRequired(boolean irr) {
		isReadRequired = irr;
	}

	/**
	 * Return the flag saying that the server is sending a body
	 */
	public boolean isReadRequired() {
		return isReadRequired;
	}

	/**
	 * Must be called before or after each request
	 */
	public void recycle() {
		if (isReadRequired && isFirst) {
			// The Servlet never read the request body, so we need to junk it
			try {
				receive();
			} catch (IOException iex) {
				log.debug("Error consuming request body", iex);
			}
		}

		end_of_stream = false;
		isEmpty = true;
		isFirst = true;
		isReplay = false;
		isReadRequired = false;
		bodyBuff.recycle();
		tempMB.recycle();
	}

	public void endMessage() throws IOException {
		outputMsg.reset();
		outputMsg.appendByte(AjpConstants.JK_AJP13_END_RESPONSE);
		outputMsg.appendByte(1);
		mc.getSource().send(outputMsg, mc);
		mc.getSource().flush(outputMsg, mc);
	}

	// -------------------- OutputBuffer implementation --------------------

	public int doWrite(ByteChunk chunk, Response res) throws IOException {
		if (!res.isCommitted()) {
			// Send the connector a request for commit. The connector should
			// then validate the headers, send them (using sendHeader) and
			// set the filters accordingly.
			res.sendHeaders();
		}

		int len = chunk.getLength();
		byte buf[] = outputMsg.getBuffer();
		// 4 - hardcoded, byte[] marshalling overhead
		int chunkSize = buf.length - outputMsg.getHeaderLength() - 4;
		int off = 0;
		while (len > 0) {
			int thisTime = len;
			if (thisTime > chunkSize) {
				thisTime = chunkSize;
			}
			len -= thisTime;

			outputMsg.reset();
			outputMsg.appendByte(AjpConstants.JK_AJP13_SEND_BODY_CHUNK);
			if (log.isTraceEnabled())
				log.trace("doWrite " + off + " " + thisTime + " " + len);
			outputMsg.appendBytes(chunk.getBytes(), chunk.getOffset() + off,
					thisTime);
			off += thisTime;
			mc.getSource().send(outputMsg, mc);
		}
		return 0;
	}

	public int doRead(ByteChunk responseChunk, Request req) throws IOException {

		if (log.isDebugEnabled())
			log.debug("doRead " + end_of_stream + " "
					+ responseChunk.getOffset() + " "
					+ responseChunk.getLength());
		if (end_of_stream) {
			return -1;
		}

		if (isFirst && isReadRequired) {
			// Handle special first-body-chunk, but only if httpd expects it.
			if (!receive()) {
				return 0;
			}
		} else if (isEmpty) {
			if (!refillReadBuffer()) {
				return -1;
			}
		}
		ByteChunk bc = bodyBuff.getByteChunk();
		responseChunk.setBytes(bc.getBuffer(), bc.getStart(), bc.getLength());
		isEmpty = true;
		return responseChunk.getLength();
	}

	/**
	 * Receive a chunk of data. Called to implement the 'special' packet in
	 * ajp13 and to receive the data after we send a GET_BODY packet
	 */
	public boolean receive() throws IOException {
		isFirst = false;
		bodyMsg.reset();
		int err = mc.getSource().receive(bodyMsg, mc);
		if (log.isDebugEnabled())
			log.info("Receiving: getting request body chunk " + err + " "
					+ bodyMsg.getLen());

		if (err < 0) {
			throw new IOException();
		}

		// No data received.
		if (bodyMsg.getLen() == 0) { // just the header
			// Don't mark 'end of stream' for the first chunk.
			// end_of_stream = true;
			return false;
		}
		int blen = bodyMsg.peekInt();

		if (blen == 0) {
			return false;
		}

		if (log.isTraceEnabled()) {
			bodyMsg.dump("Body buffer");
		}

		bodyMsg.getBytes(bodyBuff);
		if (log.isTraceEnabled())
			log.trace("Data:\n" + bodyBuff);
		isEmpty = false;
		return true;
	}

	/**
	 * Get more request body data from the web server and store it in the
	 * internal buffer.
	 * 
	 * @return true if there is more data, false if not.
	 */
	private boolean refillReadBuffer() throws IOException {
		// If the server returns an empty packet, assume that that end of
		// the stream has been reached (yuck -- fix protocol??).
		if (isReplay) {
			end_of_stream = true; // we've read everything there is
		}
		if (end_of_stream) {
			if (log.isDebugEnabled())
				log.debug("refillReadBuffer: end of stream ");
			return false;
		}

		// Why not use outBuf??
		bodyMsg.reset();
		bodyMsg.appendByte(AjpConstants.JK_AJP13_GET_BODY_CHUNK);
		bodyMsg.appendInt(AjpConstants.MAX_READ_SIZE);

		if (log.isDebugEnabled())
			log.debug("refillReadBuffer " + Thread.currentThread());

		mc.getSource().send(bodyMsg, mc);
		mc.getSource().flush(bodyMsg, mc); // Server needs to get it

		// In JNI mode, response will be in bodyMsg. In TCP mode, response need
		// to be
		// read

		boolean moreData = receive();
		if (!moreData) {
			end_of_stream = true;
		}
		return moreData;
	}

	public void appendHead(Response res) throws IOException {
		if (log.isDebugEnabled())
			log.debug("COMMIT sending headers " + res + " "
					+ res.getMimeHeaders());

		C2BConverter c2b = mc.getConverter();

		outputMsg.reset();
		outputMsg.appendByte(AjpConstants.JK_AJP13_SEND_HEADERS);
		outputMsg.appendInt(res.getStatus());

		String message = null;
		if (org.apache.coyote.Constants.USE_CUSTOM_STATUS_MSG_IN_HEADER) {
			message = res.getMessage();
		}
		if (message == null) {
			message = HttpMessages.getMessage(res.getStatus());
		} else {
			message = message.replace('\n', ' ').replace('\r', ' ');
		}
		tempMB.setString(message);
		c2b.convert(tempMB);
		outputMsg.appendBytes(tempMB);

		// XXX add headers

		MimeHeaders headers = res.getMimeHeaders();
		String contentType = res.getContentType();
		if (contentType != null) {
			headers.setValue("Content-Type").setString(contentType);
		}
		String contentLanguage = res.getContentLanguage();
		if (contentLanguage != null) {
			headers.setValue("Content-Language").setString(contentLanguage);
		}
		long contentLength = res.getContentLengthLong();
		if (contentLength >= 0) {
			headers.setValue("Content-Length").setLong(contentLength);
		}
		int numHeaders = headers.size();
		outputMsg.appendInt(numHeaders);
		for (int i = 0; i < numHeaders; i++) {
			MessageBytes hN = headers.getName(i);
			// no header to sc conversion - there's little benefit
			// on this direction
			c2b.convert(hN);
			outputMsg.appendBytes(hN);

			MessageBytes hV = headers.getValue(i);
			c2b.convert(hV);
			outputMsg.appendBytes(hV);
		}
		mc.getSource().send(outputMsg, mc);
	}

	/**
	 * Set the replay buffer for Form auth
	 */
	public void setReplay(ByteChunk replay) {
		isFirst = false;
		isEmpty = false;
		isReplay = true;
		bodyBuff.setBytes(replay.getBytes(), replay.getStart(), replay
				.getLength());
	}

}
